package com.boarsoft.boar.batch.service.impl;

import java.util.Date;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.boarsoft.boar.batch.BatchContext;
import com.boarsoft.boar.batch.biz.BatchEventBiz;
import com.boarsoft.boar.batch.biz.BatchExecutionBiz;
import com.boarsoft.boar.batch.biz.BatchInfoBiz;
import com.boarsoft.boar.batch.entity.BatchExecution;
import com.boarsoft.boar.batch.entity.BatchInfo;
import com.boarsoft.boar.batch.service.BatchBrokerService;
import com.boarsoft.boar.batch.service.BatchDispatcher;
import com.boarsoft.boar.batch.service.BatchInspector;
import com.boarsoft.common.Util;
import com.boarsoft.common.util.InetUtil;
import com.boarsoft.common.util.DateUtil;
import com.boarsoft.rpc.core.RpcContext;

/**
 * 实现对运行中的批量进行监控和干预<br>
 * 此类被BatchExecutionQuartzJob定时调起，不存在分布式并发情形
 * 
 * @author Mac_J
 *
 */
@Component("batchInspector")
public class BatchInspectorImpl implements BatchInspector {
	private final static Logger log = LoggerFactory.getLogger(BatchInspectorImpl.class);

	@Autowired
	protected BatchExecutionBiz executionBiz;
	@Autowired
	protected BatchInfoBiz infoBiz;
	// @Autowired
	// protected BatchScheduler batchScheduler;
	/** 负责执行批量的远程接口 */
	// @Autowired
	protected BatchBrokerService broker;
	/** */
	@Autowired
	protected BatchEventBiz eventBiz;

	@Override
	public void resume() {
		log.debug("Check suspending batch executions");
		Date now = new Date();
		// 查询挂起状态的任务（前面已将超时的任务状态改为超时）
		List<BatchExecution> exeLt = executionBiz.list(BatchExecution.STATUS_SUSPEND);
		if (exeLt == null || exeLt.isEmpty()) {
			return;
		}
		for (BatchExecution exe : exeLt) {
			BatchInfo info = infoBiz.get(exe.getInfoId());
			if (info == null) {
				continue;
			}
			// 检查当前任务的续跑的时间间隔
			Date st = Util.str2date(exe.getStartTime(), Util.STDDTMF);
			Date d = DateUtil.addMinute(st, info.getPeriod());
			if (d.after(now)) {
				continue;
			}
			log.info("Resume batch execution {}", exe);
			try {
				String impl = info.getImpl();
				BatchDispatcher dispatcher = BatchContext.getDispatcher(impl);
				if(dispatcher instanceof BatchDispatcherImpl) {
					BatchBrokerService broker = ((BatchDispatcherImpl) dispatcher).getBroker();
					RpcContext.specify2(exe.getAddr());
					try {
						broker.resume(exe);
					} finally {
						RpcContext.specify2(null);
					}
					log.info("dispatcher is {}, addr is {}", impl, exe.getAddr());
				}
			} catch (Exception e) {
				log.error("Error on resume batch execution {}", exe, e);
			}
		}
	}

	@Override
	public void timeout() {
		log.debug("Check timeout batch executions");
		// 查看运行超时的 Step / Job / Group，时间单位分钟
		Date now = new Date();
		List<BatchExecution> exeLt = executionBiz.list();
		for (BatchExecution exe : exeLt) {
			BatchInfo info = infoBiz.get(exe.getInfoId());
			if (info == null) {
				continue;
			}
			Date st = Util.str2date(exe.getStartTime(), Util.STDDTMF);
			Date d = DateUtil.addMinute(st, info.getTimeout());
			if (d.after(now)) {
				continue;
			}
			// 强制停止此任务
			try {
				String impl = info.getImpl();
				BatchDispatcher dispatcher = BatchContext.getDispatcher(impl);
				if(dispatcher instanceof BatchDispatcherImpl) {
					BatchBrokerService broker = ((BatchDispatcherImpl) dispatcher).getBroker();
					RpcContext.specify2(exe.getAddr());
					broker.cancel(exe,true);
					log.info("dispatcher is {}, addr is {}", impl, exe.getAddr());
				}
			} catch (Exception e) {
				log.error("Error on cancel batch execution {} by force", exe, e);
			}
			log.warn("Batch execution {} timeout.", exe);
			exe.setStatus(BatchExecution.STATUS_FAILED);
			exe.setError("timeout");
			exe.setAddr(InetUtil.getAddr());
			executionBiz.save(exe);
		}
	}

	public BatchExecutionBiz getExecutionBiz() {
		return executionBiz;
	}

	public void setExecutionBiz(BatchExecutionBiz executionBiz) {
		this.executionBiz = executionBiz;
	}

	public BatchInfoBiz getInfoBiz() {
		return infoBiz;
	}

	public void setInfoBiz(BatchInfoBiz infoBiz) {
		this.infoBiz = infoBiz;
	}

	public BatchBrokerService getBroker() {
		return broker;
	}

	public void setBroker(BatchBrokerService broker) {
		this.broker = broker;
	}

	public BatchEventBiz getEventBiz() {
		return eventBiz;
	}

	public void setEventBiz(BatchEventBiz eventBiz) {
		this.eventBiz = eventBiz;
	}
}
